﻿using Confluent.Kafka;
using Confluent.Kafka.Admin;
using KafKa.SchoolApi.Model;
using Microsoft.AspNetCore.Mvc;
using Newtonsoft.Json;
using RabbitMQ.Client;
using System.Text;

namespace KafKa.SchoolApi.Controllers 
{

    /// <summary>
    /// 
    /// </summary>
    [ApiController]
    [Route("kafka")]
    public class KafkaController : ControllerBase
    {
        private readonly ILogger<SchoolNoticeController> _logger;
        /// <summary>
        /// 
        /// </summary>
        /// <param name="logger"></param>
        public KafkaController(ILogger<SchoolNoticeController> logger)
        {
            _logger = logger;
        }

        /// <summary>
        /// 创建分区(更新分区)
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="PartitionCount"></param>
        /// <returns></returns>
        [HttpGet("PartitionUpdate")]
        public async Task PartitionCreate(string topic, int PartitionCount)
        {
            AdminClientConfig adminClientConfig = new AdminClientConfig
            {
                BootstrapServers = "127.0.0.1:9092"
            };

            var builder = new AdminClientBuilder(adminClientConfig).Build();
            builder.CreatePartitionsAsync(new PartitionsSpecification[]
            {
                new PartitionsSpecification{Topic=topic,IncreaseTo=PartitionCount},
            }).Wait();
            await Task.CompletedTask;
        }

        /// <summary>
        /// 创建主题-Kafka是一个主题（Topics）对应多个分区（Partitions），
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="Partitions"></param>
        /// <returns></returns>
        [HttpGet("TopicCreate")]
        public async Task TopicCreate(string topic)
        {
            AdminClientConfig adminClientConfig = new AdminClientConfig
            {
                BootstrapServers = "127.0.0.1:9092",
            };

            var bu = new AdminClientBuilder(adminClientConfig).Build();
            bu.CreateTopicsAsync(new TopicSpecification[] {
                    new TopicSpecification { Name = topic}
                }).Wait();

            await Task.CompletedTask;
        }

        /// <summary>
        /// 创建主题和分区
        /// </summary>
        /// <param name="topic"></param>
        /// <param name="Partitions"></param>
        /// <returns></returns>
        [HttpGet("TopicPartitionCreate")]
        public async Task TopicPartitionCreate(string topic, int PartitionCount)
        {
            AdminClientConfig adminClientConfig = new AdminClientConfig
            {
                BootstrapServers = "127.0.0.1:9092",
            };

            var bu = new AdminClientBuilder(adminClientConfig).Build();
            bu.CreateTopicsAsync(new TopicSpecification[] {
                    new TopicSpecification { Name = topic,NumPartitions =PartitionCount}
                }).Wait();

            await Task.CompletedTask;
        }

        /// <summary>
        /// 1、生产者-固定分区发送
        /// </summary>
        /// <param name="schoolNotices"></param>
        /// <returns></returns>
        [HttpPost("noticesCreate1")]
        public SchoolNotices CreateNotice1(SchoolNotices schoolNotices)
        {

            #region 生产者-固定分区发送
            {
                var producerConfig = new ProducerConfig
                {
                    BootstrapServers = "127.0.0.1:9092",
                    MessageTimeoutMs = 50000
                };
                var builder = new ProducerBuilder<string, string>(producerConfig);
                using (var producer = builder.Build())
                {
                    try
                    {
                        var teacherJson = JsonConvert.SerializeObject(schoolNotices);
                        //TopicPartition topicPartition = new TopicPartition("teacher-create", new Partition(0));
                        var dr = producer.ProduceAsync("teacher-create", new Message<string, string> { Key = "teacher", Value = teacherJson }).GetAwaiter().GetResult();
                        _logger.LogInformation("发送事件 {0} 到 {1} 成功", dr.Value, dr.TopicPartitionOffset);
                    }
                    catch (ProduceException<string, string> ex)
                    {
                        _logger.LogError(ex, "发送事件到 {0} 失败，原因 {1} ", "teacher", ex.Error.Reason);
                    }
                }
            }
            #endregion

            return null;
        }

        /// <summary>
        /// 2、生产者-失败重试-多消息发送
        /// </summary>
        /// <param name="schoolNotices"></param>
        /// <returns></returns>
        [HttpPost("noticesCreate2")]
        public SchoolNotices CreateNotice2(SchoolNotices schoolNotices)
        {

            #region 生产者-失败重试-多消息发送
            {
                var producerConfig = new ProducerConfig
                {
                    BootstrapServers = "127.0.0.1:9092",
                    MessageTimeoutMs = 50000,
                    EnableIdempotence = true,
                    TransactionalId = Guid.NewGuid().ToString()
                };

                var builder = new ProducerBuilder<string, string>(producerConfig);
                using (var producer = builder.Build())
                {
                    // 1、初始化事务
                    producer.InitTransactions(TimeSpan.FromSeconds(60));
                    try
                    {
                        var schoolNoticeJson = JsonConvert.SerializeObject(schoolNotices);
                        // 2、开发事务
                        producer.BeginTransaction();
                        for (int i = 0; i < 100; i++)
                        {
                            var dr = producer.ProduceAsync("school-notice-create", new Message<string, string> { Key = "schoolNotice", Value = schoolNoticeJson }).GetAwaiter().GetResult();
                            _logger.LogInformation("发送事件 {0} 到 {1} 成功", dr.Value, dr.TopicPartitionOffset);
                        }
                        // 3、提交事务
                        producer.CommitTransaction();
                    }
                    catch (ProduceException<string, string> ex)
                    {
                        _logger.LogError(ex, "发送事件到 {0} 失败，原因 {1} ", "user", ex.Error.Reason);
                        // 4、关闭事务
                        producer.AbortTransaction();
                    }
                }
            }
            #endregion

            return null;
        }

    }
}
